package storage

import (
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sync"
	"time"

	"github.com/louloulin/dataflare/pkg/processing/data"
)

// StorageOptions 表示存储选项
type StorageOptions struct {
	// MaxFileSize 是最大文件大小
	MaxFileSize int64
	// MaxSegments 是最大段数
	MaxSegments int
	// FlushInterval 是刷新间隔
	FlushInterval time.Duration
	// SyncOnWrite 是写入时同步
	SyncOnWrite bool
	// CompactionThreshold 是压缩阈值
	CompactionThreshold float64
}

// DefaultStorageOptions 返回默认存储选项
func DefaultStorageOptions() *StorageOptions {
	return &StorageOptions{
		MaxFileSize:         1024 * 1024 * 100, // 100MB
		MaxSegments:         10,
		FlushInterval:       time.Second * 5,
		SyncOnWrite:         false,
		CompactionThreshold: 0.5,
	}
}

// StorageEngine 表示存储引擎接口
type StorageEngine interface {
	// Open 打开存储
	Open(ctx context.Context) error
	// Close 关闭存储
	Close(ctx context.Context) error
	// Write 写入记录
	Write(ctx context.Context, record *data.DataRecord) error
	// WriteBatch 写入批次
	WriteBatch(ctx context.Context, batch *data.DataBatch) error
	// Read 读取记录
	Read(ctx context.Context, key []byte) (*data.DataRecord, error)
	// Scan 扫描记录
	Scan(ctx context.Context, startKey, endKey []byte, limit int) ([]*data.DataRecord, error)
	// Delete 删除记录
	Delete(ctx context.Context, key []byte) error
	// Flush 刷新存储
	Flush(ctx context.Context) error
	// Compact 压缩存储
	Compact(ctx context.Context) error
}

// FileSegment 表示文件段
type FileSegment struct {
	// ID 是段ID
	ID int
	// Path 是文件路径
	Path string
	// File 是文件句柄
	File *os.File
	// Size 是文件大小
	Size int64
	// CreatedAt 是创建时间
	CreatedAt time.Time
	// LastModifiedAt 是最后修改时间
	LastModifiedAt time.Time
	// Sealed 表示是否已封存
	Sealed bool
	// Index 是索引
	Index map[string]int64
	// mu 是互斥锁
	mu sync.RWMutex
}

// NewFileSegment 创建一个新的文件段
func NewFileSegment(id int, path string) *FileSegment {
	return &FileSegment{
		ID:             id,
		Path:           path,
		Size:           0,
		CreatedAt:      time.Now(),
		LastModifiedAt: time.Now(),
		Sealed:         false,
		Index:          make(map[string]int64),
	}
}

// Open 打开文件段
func (s *FileSegment) Open() error {
	s.mu.Lock()
	defer s.mu.Unlock()

	// 创建目录
	dir := filepath.Dir(s.Path)
	err := os.MkdirAll(dir, 0755)
	if err != nil {
		return fmt.Errorf("failed to create directory: %w", err)
	}

	// 打开文件
	file, err := os.OpenFile(s.Path, os.O_RDWR|os.O_CREATE, 0644)
	if err != nil {
		return fmt.Errorf("failed to open file: %w", err)
	}
	s.File = file

	// 获取文件大小
	info, err := file.Stat()
	if err != nil {
		file.Close()
		return fmt.Errorf("failed to get file info: %w", err)
	}
	s.Size = info.Size()

	// 加载索引
	err = s.loadIndex()
	if err != nil {
		file.Close()
		return fmt.Errorf("failed to load index: %w", err)
	}

	return nil
}

// Close 关闭文件段
func (s *FileSegment) Close() error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.File != nil {
		err := s.File.Close()
		s.File = nil
		return err
	}
	return nil
}

// Write 写入记录
func (s *FileSegment) Write(record *data.DataRecord, syncOnWrite bool) (int64, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.Sealed {
		return 0, fmt.Errorf("segment is sealed")
	}

	if s.File == nil {
		return 0, fmt.Errorf("segment is not open")
	}

	// 获取当前位置
	offset, err := s.File.Seek(0, io.SeekEnd)
	if err != nil {
		return 0, fmt.Errorf("failed to seek to end: %w", err)
	}

	// 序列化记录
	data, err := serializeRecord(record)
	if err != nil {
		return 0, fmt.Errorf("failed to serialize record: %w", err)
	}

	// 写入记录
	n, err := s.File.Write(data)
	if err != nil {
		return 0, fmt.Errorf("failed to write record: %w", err)
	}

	// 同步文件
	if syncOnWrite {
		err = s.File.Sync()
		if err != nil {
			return 0, fmt.Errorf("failed to sync file: %w", err)
		}
	}

	// 更新索引
	key := string(record.Key)
	s.Index[key] = offset

	// 更新大小和修改时间
	s.Size += int64(n)
	s.LastModifiedAt = time.Now()

	return offset, nil
}

// Read 读取记录
func (s *FileSegment) Read(key []byte) (*data.DataRecord, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	if s.File == nil {
		return nil, fmt.Errorf("segment is not open")
	}

	// 查找索引
	offset, ok := s.Index[string(key)]
	if !ok {
		return nil, fmt.Errorf("key not found")
	}

	// 定位到记录
	_, err := s.File.Seek(offset, io.SeekStart)
	if err != nil {
		return nil, fmt.Errorf("failed to seek to record: %w", err)
	}

	// 读取记录
	record, err := deserializeRecord(s.File)
	if err != nil {
		return nil, fmt.Errorf("failed to deserialize record: %w", err)
	}

	return record, nil
}

// Scan 扫描记录
func (s *FileSegment) Scan(startKey, endKey []byte, limit int) ([]*data.DataRecord, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	if s.File == nil {
		return nil, fmt.Errorf("segment is not open")
	}

	// 构建键范围
	startKeyStr := string(startKey)
	endKeyStr := string(endKey)

	// 收集匹配的键
	keys := make([]string, 0)
	for key := range s.Index {
		if (len(startKeyStr) == 0 || key >= startKeyStr) && (len(endKeyStr) == 0 || key < endKeyStr) {
			keys = append(keys, key)
		}
	}

	// 限制结果数量
	if limit > 0 && len(keys) > limit {
		keys = keys[:limit]
	}

	// 读取记录
	records := make([]*data.DataRecord, 0, len(keys))
	for _, key := range keys {
		offset := s.Index[key]

		// 定位到记录
		_, err := s.File.Seek(offset, io.SeekStart)
		if err != nil {
			return nil, fmt.Errorf("failed to seek to record: %w", err)
		}

		// 读取记录
		record, err := deserializeRecord(s.File)
		if err != nil {
			return nil, fmt.Errorf("failed to deserialize record: %w", err)
		}

		records = append(records, record)
	}

	return records, nil
}

// Seal 封存文件段
func (s *FileSegment) Seal() error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.Sealed {
		return nil
	}

	if s.File != nil {
		// 同步文件
		err := s.File.Sync()
		if err != nil {
			return fmt.Errorf("failed to sync file: %w", err)
		}
	}

	s.Sealed = true
	return nil
}

// loadIndex 加载索引
func (s *FileSegment) loadIndex() error {
	if s.File == nil {
		return fmt.Errorf("segment is not open")
	}

	// 重置文件指针
	_, err := s.File.Seek(0, io.SeekStart)
	if err != nil {
		return fmt.Errorf("failed to seek to start: %w", err)
	}

	// 清空索引
	s.Index = make(map[string]int64)

	// 读取文件并构建索引
	var offset int64 = 0
	for {
		// 记录当前位置
		pos, err := s.File.Seek(0, io.SeekCurrent)
		if err != nil {
			return fmt.Errorf("failed to get current position: %w", err)
		}
		offset = pos

		// 读取记录
		record, err := deserializeRecord(s.File)
		if err != nil {
			if errors.Is(err, io.EOF) {
				break
			}
			return fmt.Errorf("failed to deserialize record: %w", err)
		}

		// 更新索引
		key := string(record.Key)
		s.Index[key] = offset
	}

	return nil
}

// FileStorage 表示文件存储
type FileStorage struct {
	// basePath 是基础路径
	basePath string
	// options 是存储选项
	options *StorageOptions
	// segments 是段列表
	segments []*FileSegment
	// activeSegment 是活动段
	activeSegment *FileSegment
	// nextSegmentID 是下一个段ID
	nextSegmentID int
	// mu 是互斥锁
	mu sync.RWMutex
	// flushTicker 是刷新定时器
	flushTicker *time.Ticker
	// ctx 是上下文
	ctx context.Context
	// cancel 是取消函数
	cancel context.CancelFunc
	// wg 是等待组
	wg sync.WaitGroup
}

// NewFileStorage 创建一个新的文件存储
func NewFileStorage(basePath string, options *StorageOptions) *FileStorage {
	if options == nil {
		options = DefaultStorageOptions()
	}

	ctx, cancel := context.WithCancel(context.Background())

	return &FileStorage{
		basePath:      basePath,
		options:       options,
		segments:      make([]*FileSegment, 0),
		nextSegmentID: 1,
		ctx:           ctx,
		cancel:        cancel,
	}
}

// Open 打开存储
func (s *FileStorage) Open(ctx context.Context) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	// 创建基础目录
	err := os.MkdirAll(s.basePath, 0755)
	if err != nil {
		return fmt.Errorf("failed to create base directory: %w", err)
	}

	// 加载现有段
	err = s.loadSegments()
	if err != nil {
		return fmt.Errorf("failed to load segments: %w", err)
	}

	// 创建活动段
	if len(s.segments) == 0 {
		err = s.createNewSegment()
		if err != nil {
			return fmt.Errorf("failed to create new segment: %w", err)
		}
	} else {
		// 使用最后一个段作为活动段
		s.activeSegment = s.segments[len(s.segments)-1]
		// 如果活动段已封存或已满，创建新段
		if s.activeSegment.Sealed || s.activeSegment.Size >= s.options.MaxFileSize {
			err = s.createNewSegment()
			if err != nil {
				return fmt.Errorf("failed to create new segment: %w", err)
			}
		}
	}

	// 启动刷新定时器
	s.flushTicker = time.NewTicker(s.options.FlushInterval)
	s.wg.Add(1)
	go s.flushLoop()

	return nil
}

// Close 关闭存储
func (s *FileStorage) Close(ctx context.Context) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	// 取消上下文
	s.cancel()

	// 停止刷新定时器
	if s.flushTicker != nil {
		s.flushTicker.Stop()
	}

	// 等待刷新循环结束
	s.wg.Wait()

	// 刷新活动段
	if s.activeSegment != nil {
		err := s.activeSegment.Seal()
		if err != nil {
			return fmt.Errorf("failed to seal active segment: %w", err)
		}
	}

	// 关闭所有段
	for _, segment := range s.segments {
		err := segment.Close()
		if err != nil {
			return fmt.Errorf("failed to close segment: %w", err)
		}
	}

	return nil
}

// Write 写入记录
func (s *FileStorage) Write(ctx context.Context, record *data.DataRecord) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	// 检查活动段
	if s.activeSegment == nil {
		return fmt.Errorf("no active segment")
	}

	// 如果活动段已封存或已满，创建新段
	if s.activeSegment.Sealed || s.activeSegment.Size >= s.options.MaxFileSize {
		err := s.createNewSegment()
		if err != nil {
			return fmt.Errorf("failed to create new segment: %w", err)
		}
	}

	// 写入记录
	_, err := s.activeSegment.Write(record, s.options.SyncOnWrite)
	if err != nil {
		return fmt.Errorf("failed to write record: %w", err)
	}

	return nil
}

// WriteBatch 写入批次
func (s *FileStorage) WriteBatch(ctx context.Context, batch *data.DataBatch) error {
	for _, record := range batch.Records {
		err := s.Write(ctx, record)
		if err != nil {
			return fmt.Errorf("failed to write record: %w", err)
		}
	}
	return nil
}

// Read 读取记录
func (s *FileStorage) Read(ctx context.Context, key []byte) (*data.DataRecord, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	// 从最新的段开始查找
	for i := len(s.segments) - 1; i >= 0; i-- {
		segment := s.segments[i]
		record, err := segment.Read(key)
		if err == nil {
			return record, nil
		}
		// 如果不是"键未找到"错误，则返回错误
		if err.Error() != "key not found" {
			return nil, fmt.Errorf("failed to read from segment %d: %w", segment.ID, err)
		}
	}

	return nil, fmt.Errorf("key not found")
}

// Scan 扫描记录
func (s *FileStorage) Scan(ctx context.Context, startKey, endKey []byte, limit int) ([]*data.DataRecord, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	// 收集所有段的记录
	allRecords := make([]*data.DataRecord, 0)
	remainingLimit := limit

	// 从最新的段开始扫描
	for i := len(s.segments) - 1; i >= 0; i-- {
		segment := s.segments[i]
		records, err := segment.Scan(startKey, endKey, remainingLimit)
		if err != nil {
			return nil, fmt.Errorf("failed to scan segment %d: %w", segment.ID, err)
		}

		allRecords = append(allRecords, records...)

		// 更新剩余限制
		if limit > 0 {
			remainingLimit -= len(records)
			if remainingLimit <= 0 {
				break
			}
		}
	}

	return allRecords, nil
}

// Delete 删除记录
func (s *FileStorage) Delete(ctx context.Context, key []byte) error {
	// 创建一个墓碑记录
	tombstone := data.NewDataRecord(key, nil, data.DataFormatBinary)
	tombstone.SetHeader("tombstone", "true")

	// 写入墓碑记录
	return s.Write(ctx, tombstone)
}

// Flush 刷新存储
func (s *FileStorage) Flush(ctx context.Context) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.activeSegment != nil {
		// 同步文件
		if s.activeSegment.File != nil {
			err := s.activeSegment.File.Sync()
			if err != nil {
				return fmt.Errorf("failed to sync active segment: %w", err)
			}
		}
	}

	return nil
}

// Compact 压缩存储
func (s *FileStorage) Compact(ctx context.Context) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	// 检查是否需要压缩
	if len(s.segments) <= 1 {
		return nil
	}

	// 计算可压缩的段
	compactableSegments := make([]*FileSegment, 0)
	for i := 0; i < len(s.segments)-1; i++ {
		segment := s.segments[i]
		if segment.Sealed {
			compactableSegments = append(compactableSegments, segment)
		}
	}

	// 如果没有可压缩的段，则返回
	if len(compactableSegments) <= 1 {
		return nil
	}

	// 创建新段
	compactedSegment := NewFileSegment(s.nextSegmentID, filepath.Join(s.basePath, fmt.Sprintf("segment-%d.data", s.nextSegmentID)))
	s.nextSegmentID++

	// 打开新段
	err := compactedSegment.Open()
	if err != nil {
		return fmt.Errorf("failed to open compacted segment: %w", err)
	}

	// 收集所有键
	keys := make(map[string]bool)
	for _, segment := range compactableSegments {
		for key := range segment.Index {
			keys[key] = true
		}
	}

	// 写入最新的记录
	for key := range keys {
		record, err := s.Read(ctx, []byte(key))
		if err != nil {
			compactedSegment.Close()
			return fmt.Errorf("failed to read record: %w", err)
		}

		// 跳过墓碑记录
		if tombstone, ok := record.GetHeader("tombstone"); ok && tombstone == "true" {
			continue
		}

		// 写入记录
		_, err = compactedSegment.Write(record, false)
		if err != nil {
			compactedSegment.Close()
			return fmt.Errorf("failed to write record: %w", err)
		}
	}

	// 封存新段
	err = compactedSegment.Seal()
	if err != nil {
		compactedSegment.Close()
		return fmt.Errorf("failed to seal compacted segment: %w", err)
	}

	// 关闭可压缩的段
	for _, segment := range compactableSegments {
		err := segment.Close()
		if err != nil {
			compactedSegment.Close()
			return fmt.Errorf("failed to close segment: %w", err)
		}
	}

	// 删除可压缩的段文件
	for _, segment := range compactableSegments {
		err := os.Remove(segment.Path)
		if err != nil {
			return fmt.Errorf("failed to remove segment file: %w", err)
		}
	}

	// 更新段列表
	newSegments := make([]*FileSegment, 0)
	newSegments = append(newSegments, compactedSegment)
	for i := len(compactableSegments); i < len(s.segments); i++ {
		newSegments = append(newSegments, s.segments[i])
	}
	s.segments = newSegments

	return nil
}

// loadSegments 加载段
func (s *FileStorage) loadSegments() error {
	// 读取目录
	entries, err := os.ReadDir(s.basePath)
	if err != nil {
		if os.IsNotExist(err) {
			return nil
		}
		return fmt.Errorf("failed to read directory: %w", err)
	}

	// 查找段文件
	for _, entry := range entries {
		if entry.IsDir() {
			continue
		}

		// 检查文件名
		var id int
		_, err := fmt.Sscanf(entry.Name(), "segment-%d.data", &id)
		if err != nil {
			continue
		}

		// 创建段
		segment := NewFileSegment(id, filepath.Join(s.basePath, entry.Name()))
		segment.Sealed = true

		// 打开段
		err = segment.Open()
		if err != nil {
			return fmt.Errorf("failed to open segment: %w", err)
		}

		// 添加到段列表
		s.segments = append(s.segments, segment)

		// 更新下一个段ID
		if id >= s.nextSegmentID {
			s.nextSegmentID = id + 1
		}
	}

	return nil
}

// createNewSegment 创建新段
func (s *FileStorage) createNewSegment() error {
	// 封存当前活动段
	if s.activeSegment != nil {
		err := s.activeSegment.Seal()
		if err != nil {
			return fmt.Errorf("failed to seal active segment: %w", err)
		}
	}

	// 创建新段
	segment := NewFileSegment(s.nextSegmentID, filepath.Join(s.basePath, fmt.Sprintf("segment-%d.data", s.nextSegmentID)))
	s.nextSegmentID++

	// 打开段
	err := segment.Open()
	if err != nil {
		return fmt.Errorf("failed to open segment: %w", err)
	}

	// 添加到段列表
	s.segments = append(s.segments, segment)
	s.activeSegment = segment

	// 检查段数量
	if s.options.MaxSegments > 0 && len(s.segments) > s.options.MaxSegments {
		// 触发压缩
		go func() {
			err := s.Compact(context.Background())
			if err != nil {
				fmt.Printf("Failed to compact storage: %v\n", err)
			}
		}()
	}

	return nil
}

// flushLoop 刷新循环
func (s *FileStorage) flushLoop() {
	defer s.wg.Done()

	for {
		select {
		case <-s.ctx.Done():
			return
		case <-s.flushTicker.C:
			err := s.Flush(context.Background())
			if err != nil {
				fmt.Printf("Failed to flush storage: %v\n", err)
			}
		}
	}
}

// serializeRecord 序列化记录
func serializeRecord(record *data.DataRecord) ([]byte, error) {
	// 简单实现，实际应该使用更高效的序列化方式
	// 格式：键长度(4字节) + 值长度(4字节) + 键 + 值
	keyLen := len(record.Key)
	valueLen := len(record.Value)
	data := make([]byte, 8+keyLen+valueLen)

	// 写入键长度
	data[0] = byte(keyLen >> 24)
	data[1] = byte(keyLen >> 16)
	data[2] = byte(keyLen >> 8)
	data[3] = byte(keyLen)

	// 写入值长度
	data[4] = byte(valueLen >> 24)
	data[5] = byte(valueLen >> 16)
	data[6] = byte(valueLen >> 8)
	data[7] = byte(valueLen)

	// 写入键
	copy(data[8:], record.Key)

	// 写入值
	copy(data[8+keyLen:], record.Value)

	return data, nil
}

// deserializeRecord 反序列化记录
func deserializeRecord(reader io.Reader) (*data.DataRecord, error) {
	// 读取键长度
	keyLenBytes := make([]byte, 4)
	_, err := io.ReadFull(reader, keyLenBytes)
	if err != nil {
		return nil, err
	}
	keyLen := int(keyLenBytes[0])<<24 | int(keyLenBytes[1])<<16 | int(keyLenBytes[2])<<8 | int(keyLenBytes[3])

	// 读取值长度
	valueLenBytes := make([]byte, 4)
	_, err = io.ReadFull(reader, valueLenBytes)
	if err != nil {
		return nil, err
	}
	valueLen := int(valueLenBytes[0])<<24 | int(valueLenBytes[1])<<16 | int(valueLenBytes[2])<<8 | int(valueLenBytes[3])

	// 读取键
	key := make([]byte, keyLen)
	_, err = io.ReadFull(reader, key)
	if err != nil {
		return nil, err
	}

	// 读取值
	value := make([]byte, valueLen)
	_, err = io.ReadFull(reader, value)
	if err != nil {
		return nil, err
	}

	// 创建记录
	record := data.NewDataRecord(key, value, data.DataFormatBinary)
	return record, nil
}
